package com.frinder.official.akka.routing;

import akka.actor.ActorRef;
import akka.actor.Terminated;
import akka.routing.ActorRefRoutee;
import akka.routing.RoundRobinRoutingLogic;
import akka.routing.Routee;
import akka.routing.Router;
import com.frinder.official.spring.MyBaseActor;
import com.frinder.official.spring.SpringExtension;
import com.google.common.collect.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.List;

/**
 * Created by frinder on 2017/9/29.
 */
@Component
@Scope("prototype")
public class Master extends MyBaseActor {

    @Autowired
    private SpringExtension springExtension;

    Router router;

    @Override
    public void preStart() throws Exception {
        super.preStart();
        List<Routee> routees = Lists.newArrayList();
        for (int i = 0; i < 5; i++) {
            ActorRef r = getContext().actorOf(springExtension.props("child"));
            getContext().watch(r);
            routees.add(new ActorRefRoutee(r));
        }
        router = new Router(new RoundRobinRoutingLogic(), routees);
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Work.class, message -> {
                    router.route(message, getSender());
                })
                .match(Terminated.class, message -> {
                    router = router.removeRoutee(message.actor());
                    ActorRef r = getContext().actorOf(springExtension.props("child"));
                    getContext().watch(r);
                    router = router.addRoutee(new ActorRefRoutee(r));
                })
                .build();
    }

    public final class Work implements Serializable {
        private static final long serialVersionUID = 1L;
        public final String payload;

        public Work(String payload) {
            this.payload = payload;
        }
    }
}